// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.core.tsuid;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.io.Files;
import net.opentsdb.core.*;
import net.opentsdb.query.QueryUtil;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.Config;
import net.opentsdb.utils.enums.FamQuls;
import org.hbase.async.HBaseClient;
import org.hbase.async.KeyOnlyFilter;
import org.hbase.async.KeyValue;
import org.hbase.async.Scanner;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;

/**
 * 测试自定义扫描，原生扫描有毒
 */
public class TestScan {

    private static final Logger LOG = LoggerFactory.getLogger(TestScan.class);

    private static Config config;
    private static TSDB tsdb = null;
    private HBaseClient client;

    @Before
    public void before() throws Exception{
        config = new Config("C:\\user\\workspace\\project\\tsdb\\opentsdb-2.4rc2\\src-main\\net\\opentsdb\\opentsdb.conf");
        tsdb = new TSDB(config);
        client = tsdb.getClient();
    }

    @Test
    public void testScanData() throws Exception{
        final byte[] START_ROW = new byte[] { '!' };
        final byte[] END_ROW = new byte[] { '~' };

        File file = new File("C:\\Users\\zhipeng.zhu\\Documents\\shareFile\\tsuid.log");
        Files.touch( file);
//        String tsuid003 =  "00000300000100002B000002000020,000003000001000033000002000032,00000300000100002A000002000020,00000300000100002C000002000020,000003000001000036000002000032,00000300000100002E000002000020,00000300000100002D000002000020,000003000001000035000002000032,000003000001000034000002000032,000003000001000037000002000032,000003000001000030000002000020,000003000001000038000002000032,00000300000100001600000200000D,00000300000100001400000200000D,00000300000100001500000200000D,00000300000100001300000200000D,00000300000100002F000002000020,00000300000100001000000200000D,00000300000100001100000200000D,00000300000100001F000002000020,00000300000100001200000200000D,00000300000100000B000002000002,00000300000100000E00000200000D,00000300000100000A000002000002,00000300000100000F00000200000D,000003000001000022000002000020,000003000001000021000002000020,000003000001000023000002000020,000003000001000025000002000020,000003000001000024000002000020,00000300000100000C00000200000D,000003000001000029000002000020,000003000001000026000002000020,000003000001000027000002000020,000003000001000028000002000020,000003000001000003000002000002,000003000001000004000002000002,000003000001000008000002000002,000003000001000001000002000002,000003000001000009000002000002,000003000001000007000002000002,000003000001000006000002000002,000003000001000005000002000002,00000300000100001700000200000D,00000300000100001800000200000D,00000300000100001900000200000D,00000300000100001D00000200000D,00000300000100001E00000200000D,00000300000100001C00000200000D,00000300000100001A00000200000D,00000300000100001B00000200000D,000003000001000031000002000032";
        String tsuid003 =  "00000300000100002B000002000020";

        List<String> tsuids = Arrays.asList(tsuid003.split(","));
        Set<String> tsuidSet = Sets.newHashSet();
        final Scanner scanner = client.newScanner(tsdb.dataTable());
        scanner.setMaxNumRows(32);
//        scanner.setStartKey(START_ROW);
//        scanner.setStopKey(END_ROW);
//        scanner.setFamily(FamQuls.META.FAMILY.NAME.getByte());
        scanner.setFamily(TSDB.FAMILY());
//
//        scanner.setFilter(new KeyOnlyFilter());
        scanner.setKeyRegexp(QueryUtil.getRowKeyTSUIDRegex(tsuids), StandardCharsets.UTF_8);
        ArrayList<ArrayList<KeyValue>> rows;
        int count = 0;

        System.out.println("-----------------------------------start scan--------------------------");
        try{
            while ((rows = scanner.nextRows().joinUninterruptibly()) != null){
                for (final ArrayList<KeyValue> row : rows) {
//                    System.out.println("row:"+row.size());
//                    LOG.debug("row:{}",row.size());
                    if(row != null && row.size()>0){
                        final byte[] tsuid = UniqueId.getTSUIDFromKey(row.get(0).key(), TSDB.metrics_width(), Const.TIMESTAMP_BYTES);
                        String tsuidStr = UniqueId.uidToString(tsuid);
                        tsuidSet.add(tsuidStr);
                        Files.append("\n"+String.valueOf(row),file, StandardCharsets.UTF_8);
                        count += row.size();
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("--------------scan error-------------------"+e.getMessage());
        }
        System.out.println("-----------------------------------scan end--------------------------");
        System.out.println("coung tsuid point:"+count);
        System.out.println(tsuidSet);
    }

    @Test
    public void testQueryByTsuids() throws  Exception{
        List<String> tsuids = Lists.newArrayList("00000300000100002B000002000020");

        TSSubQuery subQuery = new TSSubQuery();
        subQuery.setAggregator("none");
        subQuery.setTsuids(tsuids);
        TSQuery query = new TSQuery();
        query.setQueries(Lists.newArrayList(subQuery));

        query.setStart("0");
        query.setEnd("1562553334000");
        query.validateAndSetQuery();

        LOG.debug("------------------------------------start get tsuids--------------------------------------");
        query.buildQueriesAsync(tsdb).addCallback(
                iQueries -> {
                    DataPoints[] dpsArr = iQueries[0].run();
                    for(DataPoints dps : dpsArr){
                        LOG.debug(String.valueOf(dps.getTSUIDs()));
                    }
                    return null;
                }
            );
    }
}
